package hub

import (
	"FeedXmh/app/dao"
	"FeedXmh/app/model"
	"FeedXmh/redis"
	"container/list"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gogf/gf/database/gdb"
	"github.com/gogf/gf/encoding/gbinary"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/encoding/gparser"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/net/ghttp"
	"github.com/gogf/gf/os/gtime"
	"github.com/gogf/gf/text/gstr"
	"github.com/gogf/gf/util/gconv"
	"github.com/gogf/gf/util/grand"
	"github.com/shopspring/decimal"
	"io/ioutil"
	"paco_toolkit/DMXFGW"
	"paco_toolkit/common"
	"time"
)

func CreateMqttClient(cid string, onConnet mqtt.OnConnectHandler, onConnectionLost mqtt.ConnectionLostHandler) (client mqtt.Client) {
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("ssl://%s:%d", g.Cfg().GetString("mqtt.broker_url"), g.Cfg().GetInt("mqtt.broker_port")))
	opts.SetClientID(cid)
	tlsconfig := newTLSConfig()
	opts.SetTLSConfig(tlsconfig)
	opts.Username = g.Cfg().GetString("mqtt.broker_username")
	opts.Password = g.Cfg().GetString("mqtt.broker_pwd")
	opts.SetKeepAlive(60 * time.Second)
	opts.SetAutoReconnect(true) //当网络断开后，客户端会进行重连
	//opts.SetCleanSession(false)//把配置里的 cleanSession 设为false，客户端掉线后 服务器端不会清除session
	opts.ResumeSubs = true //SetResumeSubs将在连接时启用已存储(un)订阅消息的恢复功能，但如果CleanSession为false，则不会重新连接。
	// 设置消息回调处理函数
	//opts.SetDefaultPublishHandler(f)
	opts.OnConnect = onConnet
	opts.OnConnectionLost = onConnectionLost

	opts.SetPingTimeout(10 * time.Second)

	//opts.SetWill("offline",opts.ClientID+" offline",0,true)
	//opts.WillEnabled=true

	client = mqtt.NewClient(opts)
	return client
}
func newTLSConfig() *tls.Config {
	// Import trusted certificates from CAfile.pem.
	// Alternatively, manually add CA certificates to
	// default openssl CA bundle.
	certpool := x509.NewCertPool()

	pemCerts, err := ioutil.ReadFile("/file/MyRootCA.crt")
	if err == nil {
		certpool.AppendCertsFromPEM(pemCerts)
	}

	// Import client certificate/key pair
	cert, err := tls.LoadX509KeyPair("file/client.crt", "file/client.key")
	if err != nil {
		panic(err)
	}

	// Just to print out the client certificate..
	cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
	if err != nil {
		panic(err)
	}
	//fmt.Println(cert.Leaf)
	//fmt.Println("NotAfter", cert.Leaf.NotAfter.String())

	// Create tls.Config with desired tls properties
	return &tls.Config{
		// RootCAs = certs used to verify server cert.
		RootCAs: certpool,
		// ClientAuth = whether to request cert from server.
		// Since the server is set up for SSL, this happens
		// anyways.
		ClientAuth: tls.RequireAndVerifyClientCert,
		// ClientCAs = certs used to validate client cert.
		ClientCAs: nil,
		// InsecureSkipVerify = verify that cert contents
		// match server. IP matches what is in cert etc.
		InsecureSkipVerify: true,
		// Certificates = list of certs client sends to server.
		Certificates: []tls.Certificate{cert},
	}
}

func GetApiClients(params gdb.Map) (res []map[string]interface{}, count int) {
	client := ghttp.NewClient()
	client.SetBasicAuth(g.Cfg().GetString("mqtt.interface_user"), g.Cfg().GetString("mqtt.interface_pwd"))
	//client.SetContentType("none")
	url := g.Cfg().GetString("mqtt.interface_url") + ghttp.BuildParams(params)
	//g.Log().Line(true).Debugf("url:%s", url)
	resp, err := client.SetTimeout(20*time.Second).Get(url, nil)
	if err == nil {
		if resp.StatusCode == 200 {
			respStr := resp.ReadAllString()
			if json, err := gjson.DecodeToJson(respStr); err == nil {
				res = json.GetMaps("data")
				count = json.GetInt("meta.count")
			} else {
				g.Log().Line().Error(err)
			}
		} else {
			respStr := resp.ReadAllString()
			g.Log().Line(true).Debug("respStr : ", respStr, "resp.StatusCode:", resp.StatusCode)
		}
	}
	resp.Close() //要手动关闭，不然会内存泄漏
	return res, count
}

// ClearNodeStatus 删除无效饲喂器状态（参数不存在，状态存在的）
func ClearNodeStatus() {
	//SELECT s.* FROM t_hzfgw_node_status s
	//LEFT JOIN t_hzfgw_node_para p ON p.sn = s.sn
	//WHERE IFNULL(p.sn,'')=''

	db := g.DB(g.Cfg().GetString("custom.db_name"))
	sql := "SELECT s.* FROM t_hzfgw_node_status s LEFT JOIN t_hzfgw_node_para p ON p.sn = s.sn WHERE IFNULL(p.sn,'')='' "
	status, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
	} else {
		list := status.List()
		if len(list) > 0 {
			for _, v := range list {
				sn := gconv.String(v["sn"])
				db.Model("t_hzfgw_node_status").Where(fmt.Sprintf("sn='%s'", sn)).Delete()
			}
		}

	}
}

// ClearInvalidAlarm 清除无效告警
func ClearInvalidAlarm() {
	month_ago := gtime.Now().AddDate(0, -1, 0).Format(common.DATE_FORMAT)
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	sql := fmt.Sprintf("SELECT * FROM t_alarm WHERE `status`=1 AND `level`=1 AND add_time<'%s' ;", month_ago)
	data, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
	} else {
		list := data.List()
		if len(list) > 0 {
			/*for _, v := range list {

				farmid := gconv.String(v["farmid"])
				houseid := gconv.String(v["houseid"])
				device_addr := gconv.String(v["device_addr"])

				device_type := gconv.String(v["device_type"])
				sn := gconv.String(v["sn"])
				remark := gconv.String(v["remark"])


			}*/
		}
	}
}

// UpdateNMSWQGWPreFeed 新版饲喂器 01:00 把昨日所有t_hzfgw_node_status_his的饲喂量更新进入redis
func UpdateNMSWQGWPreFeed() {
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	yesterday := gtime.Now().AddDate(0, 0, -1).Format(common.DATE_FORMAT)
	res, err := db.GetAll(fmt.Sprintf(" SELECT addr,sn,max_feed,(day_use_feed_man+day_use_feed_auto) as feed FROM t_hzfgw_node_status_his WHERE his_date='%s';", yesterday))
	if err != nil {
		g.Log().Line(true).Error(err)
	} else {
		list := res.List()
		if len(list) > 0 {
			for _, item := range list {
				addr := gconv.String(item["addr"])
				sn := gconv.String(item["sn"])
				pre_day_feed := gconv.String(item["feed"]) + "/" + gconv.String(item["max_feed"])
				redis.SetGwNodePreDayFeed(addr, sn, pre_day_feed)
			}
		}
	}
}

// UpdateHouseSummary 计算栏舍汇总数据
func UpdateHouseSummary() {
	farms, err := dao.T_farm.GetList()
	if err != nil {
		g.Log().Line(true).Error(err)
	} else {
		if len(farms) > 0 {
			db := g.DB(g.Cfg().GetString("custom.db_name"))
			yestoday := gtime.Now().AddDate(0, 0, -1).Format(common.DATE_FORMAT)

			for _, farm := range farms {
				farmid := gconv.String(farm["id"])
				sql := fmt.Sprintf(`SELECT houseid,farmid,farmname,housename,house_type,idling_limit,idling_enable,
sum(if(mode>0,pig_num,0)) as pig_num_total,
sum(if(mode>0,son,0)) as son_total,
sum(pig_num) as pig_num,
count(0) as device_num, -- 设备总数
sum(if(online=1,1,0)) as online_num, -- 总在线数
sum(if(online=0 || online is NULL,1,0)) as offline_num, -- 总离线数目
sum(if(mode=0,1,0)) as num_stop, -- 空栏设备
sum(if((alarm>0||alarm_ext>0) || (IFNULL(online,0)=0 && mode>0) ,1,0)) as alarm_device_num, -- 告警设备数，包含非空栏离线
sum(if(online=1,max_feed,0)) as max_feed,
sum(if(online=1,day_use_feed_auto,0)) as day_use_feed_auto,
sum(if(online=1,day_use_feed_man,0)) as day_use_feed_man,
sum(if(online=1,day_use_water_auto,0)) as day_use_water_auto,
sum(if(online=1,day_use_water_man,0)) as day_use_water_man,
sum(if(online=1,day_use_feed_auto+day_use_feed_man,0) ) as today_feed,
sum(if(online=1,day_use_water_auto+day_use_water_man,0)) as today_water
FROM v_hzgw_node WHERE farmid='%s' GROUP BY houseid ;`, farmid)
				res, err := db.GetAll(sql)
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					list := res.List()
					if len(list) > 0 {
						for _, item := range list {
							houseid := gconv.String(item["houseid"])

							sql = fmt.Sprintf(`SELECT sum(day_use_feed_auto+day_use_feed_man) AS feed
		,SUM(day_use_water_auto+day_use_water_man) as water
		FROM t_hzfgw_node_status_his 
		WHERE addr in (SELECT device_addr FROM v_device WHERE houseid='%s') and his_date='%s';`, houseid, yestoday)

							dt, err := db.GetOne(sql)
							if err != nil {
								g.Log().Line().Error(err)
							} else {
								data := dt.Map()
								item["yestoday_feed"] = data["feed"]
								item["yestoday_water"] = data["water"]
							}

							str, _ := gparser.VarToJsonString(item)
							redis.SetHouseSummary(houseid, str)
						}
					}
				}
			}
		}
	}
}

func UpdateFarmSummary() {
	farms, err := dao.T_farm.GetList()
	if err != nil {
		g.Log().Line(true).Error(err)
	} else {
		if len(farms) > 0 {
			db := g.DB(g.Cfg().GetString("custom.db_name"))
			yestoday := gtime.Now().AddDate(0, 0, -1).Format(common.DATE_FORMAT)

			for _, farm := range farms {
				farmid := gconv.String(farm["id"])
				sql := fmt.Sprintf(`SELECT farmid,farmname,
sum(if(mode>0,pig_num,0)) as pig_num_total,
sum(if(mode>0,son,0)) as son_total,
sum(pig_num) as pig_num,
sum(if(mode=1,pig_num,0)) as pig_num1,
sum(if(mode=2,pig_num,0)) as pig_num2,
sum(if(mode=3,pig_num,0)) as pig_num3,
sum(if(mode=4,pig_num,0)) as pig_num4,
sum(if(mode=5,pig_num,0)) as pig_num5,
count(0) as device_num, -- 设备总数
sum(if(online=1,1,0)) as online_num, -- 总在线数
sum(if(online=0 || online is NULL,1,0)) as offline_num, -- 总离线数目
sum(if(mode=0,1,0)) as num_stop, -- 空栏设备
sum(if((alarm>0||alarm_ext>0) || (IFNULL(online,0)=0 && mode>0) ,1,0)) as alarm_device_num, -- 告警设备数，包含非空栏离线
sum(if(online=1,max_feed,0)) as max_feed,
sum(if(online=1,day_use_feed_auto,0)) as day_use_feed_auto,
sum(if(online=1,day_use_feed_man,0)) as day_use_feed_man,
sum(if(online=1,day_use_water_auto,0)) as day_use_water_auto,
sum(if(online=1,day_use_water_man,0)) as day_use_water_man,
sum(if(online=1,day_use_feed_auto+day_use_feed_man,0) ) as today_feed,
sum(if(online=1,day_use_water_auto+day_use_water_man,0)) as today_water
FROM v_hzgw_node WHERE farmid='%s'; `, farmid)
				res, err := db.GetOne(sql)
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					item := res.Map()

					//昨日下料，昨日下水
					sql = fmt.Sprintf(`SELECT sum(day_use_feed_auto+day_use_feed_man) AS feed
		,SUM(day_use_water_auto+day_use_water_man) as water
		FROM t_hzfgw_node_status_his 
		WHERE addr in (SELECT device_addr FROM v_device WHERE farmid='%s') and his_date='%s';`, farmid, yestoday)

					dt, err := db.GetOne(sql)
					if err != nil {
						g.Log().Line(true).Error(err)
					} else {
						data := dt.Map()
						item["yestoday_feed"] = data["feed"]
						item["yestoday_water"] = data["water"]
					}

					//统计栏舍和终端
					sql = fmt.Sprintf(`SELECT count(DISTINCT houseid) as house_num,count(DISTINCT device_addr) as gw_num,sum(if(alive=1,1,0)) as gw_alive_num FROM v_device WHERE farmid='%s';`, farmid)
					dt1, err := db.GetOne(sql)
					if err != nil {
						g.Log().Line().Error(err)
					} else {
						data1 := dt1.Map()
						item["house_num"] = data1["house_num"]
						item["gw_num"] = data1["gw_num"]
						item["gw_alive_num"] = data1["gw_alive_num"]
					}

					str, _ := gparser.VarToJsonString(item)
					redis.SetFarmSummary(farmid, str)

				}
			}
		}
	}
}

/*func FindProcessDeviceByAddr(addr string) (exist bool, tryNum int) {
	exist = false
	for i := model.ProcessDeviceList.Front(); i != nil; i = i.Next() {
		t := i.Value
		d := t.(*model.ProcessDevice)
		if d.Addr == addr {
			exist = true
			tryNum = d.TryNum
			break
		}
	}
	return exist, tryNum
}

func ProcessAddCountDeviceByAddr(addr string) {
	for i := model.ProcessDeviceList.Front(); i != nil; i = i.Next() {
		t := i.Value
		d := t.(*model.ProcessDevice)
		if d.Addr == addr {
			d.TryNum += 1
			break
		}
	}
}

func ProcessDeviceResetByAddr(addr string) {
	for i := model.ProcessDeviceList.Front(); i != nil; i = i.Next() {
		t := i.Value
		d := t.(*model.ProcessDevice)
		if d.Addr == addr {
			d.TryNum = 0
			break
		}
	}
}

func RemoveProcessDeviceByAddr(addr string) (isok bool) {
	isok = false
	var n *list.Element
	for i := model.ProcessDeviceList.Front(); i != nil; i = n {
		n = i.Next()
		t := i.Value
		d := t.(*model.ProcessDevice)
		if d.Addr == addr {
			model.ProcessDeviceList.Remove(i)
			isok = true
			break
		}
	}
	return isok
}*/

func FindProcessDeviceByAddr(addr string) (exist bool) {
	exist = false
	for i := model.ProcessDeviceList.Front(); i != nil; i = i.Next() {
		t := i.Value
		if t == addr {
			exist = true
			break
		}
	}
	return exist
}
func RemoveProcessDeviceByAddr(addr string) (isok bool) {
	isok = false
	var n *list.Element
	for i := model.ProcessDeviceList.Front(); i != nil; i = n {
		n = i.Next()
		t := i.Value
		if t == addr {
			model.ProcessDeviceList.Remove(i)
			g.Log().Line(true).Infof("addr %s unlocked.", addr)
			isok = true
		}
	}
	return isok
}

func GetStandard(_model string, mode int, day int) (low float32, standard float32, high float32) {
	if _model == "feed" || _model == "" {

	} else if _model == "fg" {
		//转换成对应的mode
		if mode == DMXFGW.FG_MODE_NURSE {
			mode = DMXFGW.FEED_MODE_NURSE
		} else if mode == DMXFGW.FG_MODE_FAT {
			mode = DMXFGW.FEED_MODE_FAT
		}
	}

	if mode == DMXFGW.FEED_MODE_PG {
		low = 2.5
		standard = 3
		high = 4
	} else if mode == DMXFGW.FEED_MODE_LACTATION {
		low = 2
		standard = 3
		high = 4
	} else if mode == DMXFGW.FEED_MODE_NURSE {
		low = 0.077
		standard = 0.11
		high = 0.121
	} else if mode == DMXFGW.FEED_MODE_FAT {
		low = 0.507
		standard = 0.714
		high = 0.814
	}

	for _, v := range model.DFXW_FEED_STANDARD_DATA {
		_mode := gconv.Int(v["mode"])
		_day := gconv.Int(v["day"])

		if _mode == mode && _day == day {
			//g.Log().Line(true).Debug(v)
			low = float32(gconv.Int(v["low"])) / 1000
			standard = float32(gconv.Int(v["standard"])) / 1000
			high = float32(gconv.Int(v["high"])) / 1000
			break
		}
	}
	return low, standard, high
}

// 获取需要联动的单元
func getAllDfxwBizUnit() (list gdb.List) {
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	sql := "SELECT * FROM t_house WHERE dfxw_id>0; "
	re, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
	} else {
		list = re.List()
	}
	return list
}

// SyncAllDfxwBizData 检测环控联动到待同步记录
func SyncAllDfxwBizData() {
	g.Log().Line(true).Info("SyncAllDfxwBizData")
	houseList := getAllDfxwBizUnit()
	if len(houseList) > 0 {
		for _, house_item := range houseList {

			go func(item gdb.Map) {
				linkage_feed := gconv.Int(item["linkage_feed"])
				dfxw_id := gconv.String(item["dfxw_id"])
				//g.Log().Line(true).Infof("DFXW_getUnitBizData dfxw_id=%s", dfxw_id)
				var id_arr []string
				id_arr = append(id_arr, dfxw_id)
				biz_items, biz_time, err := DFXW_getUnitBizData(id_arr)
				if err != nil {
					g.Log().Line(true).Errorf("DFXW_getUnitBizData dfxw_id=%s err!", dfxw_id)
					g.Log().Line(true).Error(err)
					return
				}
				if len(biz_items) > 0 {
					g.Log().Line(true).Infof("DFXW_getUnitBizData dfxw_id=%s ok", dfxw_id)
					var biz_item = biz_items[0]
					var temp_change bool
					if gconv.Int(biz_item["unitAvgTemperature"]) != gconv.Int(item["dfxw_unit_avg_temperature"]) {
						//整数部分变动才变动
						temp_change = true
					} else {
						temp_change = false
					}

					item["dfxw_id"] = dfxw_id
					item["dfxw_unit_avg_temperature"] = biz_item["unitAvgTemperature"]
					item["dfxw_unit_avg_weight"] = biz_item["unitAvgWeight"]
					item["dfxw_unit_day_age"] = biz_item["unitDayAge"]
					item["dfxw_unit_humidity"] = biz_item["unitHumidity"]
					item["dfxw_unit_num"] = biz_item["unitNum"]
					item["dfxw_create_time"] = biz_item["createTime"]
					item["dfxw_update_time"] = biz_item["updateTime"]
					item["dfxw_unit_biz_timestamp"] = biz_time
					dao.T_house.Save(item)

					if temp_change && linkage_feed > 0 {
						//CheckLinkAgeFeedRecordsByUnit(house_item)
					}
				} else {
					g.Log().Line(true).Warningf("DFXW_getUnitBizData dfxw_id=%s no data!", dfxw_id)
				}
			}(house_item)

			time.Sleep(100 * time.Millisecond)

		}
	}
}

func CheckLinkAgeFeedRecordsByUnit(house_item gdb.Map) (err error) {
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	now := gtime.Now()
	houseid := gconv.Int(house_item["id"])
	housename := gconv.String(house_item["housename"])
	dfxw_unit_avg_temperature := gconv.Float32(house_item["dfxw_unit_avg_temperature"])
	dfxw_unit_humidity := gconv.Float32(house_item["dfxw_unit_humidity"])
	dfxw_unit_day_age := gconv.Int(house_item["dfxw_unit_day_age"])
	dfxw_unit_num := gconv.Int(house_item["dfxw_unit_num"])
	dfxw_unit_avg_weight := gconv.Float32(house_item["dfxw_unit_avg_weight"])
	dfxw_unit_biz_timestamp := gconv.String(house_item["dfxw_unit_biz_timestamp"])

	//g.Log().Line(true).Info("环控联动单元信息")
	g.Log().Line(true).Infof("%s,环控联动单元信息%v", housename, house_item)

	if gconv.Int(dfxw_unit_avg_temperature) <= -127 {
		err = errors.New(fmt.Sprintf("%s 单元温度无效 %.2f", housename, dfxw_unit_avg_temperature))
		g.Log().Line(true).Warningf("%s 单元温度无效 %.2f", housename, dfxw_unit_avg_temperature)
		return
	} else {
		g.Log().Line(true).Infof("%s 单元温度 %.2f", housename, dfxw_unit_avg_temperature)
	}
	if dfxw_unit_day_age <= -127 {
		err = errors.New(fmt.Sprintf("%s 单元日龄无效 %d", housename, dfxw_unit_day_age))
		g.Log().Line(true).Warningf("%s 单元日龄无效 %d", housename, dfxw_unit_day_age)
		return
	} else {
		g.Log().Line(true).Infof("%s 单元日龄 %d", housename, dfxw_unit_day_age)
	}

	if gconv.Int(dfxw_unit_humidity) <= -127 {
		g.Log().Line(true).Warningf("%s 单元湿度无效 %2f", housename, dfxw_unit_humidity)
	}
	if gconv.Int(dfxw_unit_avg_weight) <= -127 {
		g.Log().Line(true).Warningf("%s 单元均重无效 %2f", housename, dfxw_unit_avg_weight)
	}
	if dfxw_unit_num < 0 {
		g.Log().Line(true).Warningf("%s 单元猪只数量无效 %d", housename, dfxw_unit_num)
	}

	biz_timestamp := gtime.NewFromStr(dfxw_unit_biz_timestamp).TimestampMilli()
	biz_time_span := (now.TimestampMilli() - biz_timestamp) / 1000
	day_val := biz_time_span / (60 * 60 * 24)
	if day_val > 0 {
		err = errors.New(fmt.Sprintf("%s 单元业务数据时间过期 %s", housename, common.BeautyTimeSpan(biz_time_span)))
		g.Log().Line(true).Warningf("%s 单元业务数据时间过期 %s", housename, common.BeautyTimeSpan(biz_time_span))
		return
	}

	//查询单元下所有饲喂器的参数
	sql := fmt.Sprintf("SELECT addr,sn,tn,hnum,`mode`,`day`,feed_plan_id,feed_plan_name_new,`online`,para FROM v_hzgw_node WHERE houseid=%d", houseid)
	re, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
	} else {
		nodes := re.List()
		if len(nodes) > 0 {
			for _, node := range nodes {
				addr := gconv.String(node["addr"])
				sn := gconv.String(node["sn"])
				tn := gconv.String(node["tn"])
				hnum := gconv.String(node["hnum"])
				mode := gconv.Int(node["mode"])
				day := gconv.Int(node["day"])
				online := gconv.Int(node["online"])
				fpara_str := gconv.String(node["para"])
				feed_plan_id := gconv.Int(node["feed_plan_id"])
				if online == 0 {
					g.Log().Line(true).Warningf("%s 栏位:%s SN:%s 离线", housename, hnum, sn)
					//continue
				}
				if feed_plan_id == 0 {
					g.Log().Line(true).Warningf("%s 栏位:%s SN:%s 当前无方案", housename, hnum, sn)
					continue
				}

				//查询是否已经存在的待同步记录，存在则不新增
				sync_list, _ := dao.B_dfxw_linkage_feed_para.GetList(fmt.Sprintf(" sn='%s' and sync<=0 ", sn))
				if len(sync_list) > 0 {
					g.Log().Line(true).Warningf("%s 栏位:%s SN:%s 已经存在待同步记录 ", housename, hnum, sn)
					continue
				}

				//具有方案
				plan, err := dao.T_feed_plan_new.GetOne(fmt.Sprintf("id=%d", feed_plan_id))
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					plan_fpara_str := gconv.String(plan["para"])
					plan_fpara := DMXFGW.FNodePara{}
					err = gjson.DecodeTo(plan_fpara_str, &plan_fpara)
					if err != nil {
						g.Log().Line(true).Error(err)
						continue
					}
					plan_para := DMXFGW.EncodeFNodeParamToNodePara(&plan_fpara)
					if gconv.Int(plan_para.Mode) != mode {
						g.Log().Line(true).Warningf("%s 栏位:%s SN:%s 方案模式与参数模式不一致,方案模式=%d 参数模式=%d", housename, hnum, sn, plan_para.Mode, mode)
						continue
					}

					//获取方案所处日龄段的水料比
					plan_waterFeed := getParaWaterFeed(plan_para, mode, uint16(day))
					g.Log().Line(true).Infof("%s 栏位:%s SN:%s 方案水料比:%s", housename, hnum, sn, plan_waterFeed)

					offset := getWaterFeedOffset(mode, day, dfxw_unit_avg_temperature)
					g.Log().Line(true).Infof("%s 栏位:%s SN:%s 偏移值:%.1f", housename, hnum, sn, offset)

					new_waterFeed := gconv.Float32(plan_waterFeed) + offset
					g.Log().Line(true).Infof("%s 栏位:%s SN:%s 新的水料比:%.1f", housename, hnum, sn, new_waterFeed)

					fpara := DMXFGW.FNodePara{}
					err = gjson.DecodeTo(fpara_str, &fpara)
					if err != nil {
						g.Log().Line(true).Error(err)
						continue
					}
					para := DMXFGW.EncodeFNodeParamToNodePara(&fpara)
					/*isok, new_para := createParaByWaterFeed(*para, mode, day, new_waterFeed)
					if !isok {
						g.Log().Line(true).Warningf("%s 栏位:%s SN:%s 调整参数水料比值失败,未找到满足条件的日龄=%d ", housename, hnum, sn, day)
						continue
					}*/

					water_feed_idx := getParaWaterFeedAddrOffset(para, mode, day)

					//保存待调整数据到记录
					//new_fpara := DMXFGW.FormatNodeParam(&new_para)
					//new_fpara_str, _ := gparser.VarToJsonString(new_fpara)
					saveItem := gdb.Map{
						"addr":              addr,
						"sn":                sn,
						"tn":                tn,
						"hnum":              hnum,
						"mode":              mode,
						"day":               day,
						"unit_day":          dfxw_unit_day_age,
						"unit_temp":         fmt.Sprintf("%.2f", dfxw_unit_avg_temperature),
						"feed_plan_id":      feed_plan_id,
						"plan_water_feed":   plan_waterFeed,
						"target_water_feed": fmt.Sprintf("%.1f", new_waterFeed),
						//"cur_node_para":     fpara_str,
						//"target_node_para":  new_fpara_str,
						"water_feed_idx": water_feed_idx,
						"create_date":    now.String(),
						"sync":           0,
						"sync_res":       nil,
						"sync_date":      nil,
						"sync_by":        nil,
					}
					//g.Log().Line(true).Debug(saveItem)
					dao.B_dfxw_linkage_feed_para.Save(saveItem)
				}
			}
		}
	}
	return err
}

// 获取参数水料比
func getParaWaterFeed(para *DMXFGW.NodePara, mode int, day uint16) (waterFeed string) {
	waterFeed = "0"
	if mode == DMXFGW.FEED_MODE_PG {
		for k, gday := range para.Pg_gday {
			if gday <= day {
				waterFeed = gconv.String(float32(para.Pg_waterFeed[k]) / 10)
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_LACTATION {
		for k, gday := range para.La_gday {
			if gday <= day {
				waterFeed = gconv.String(float32(para.La_waterFeed[k]) / 10)
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_NURSE {
		for k, gday := range para.Nr_gday {
			if gday <= day {
				waterFeed = gconv.String(float32(para.Nr_waterFeed[k]) / 10)
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_FAT {
		for k, gday := range para.Fat_gday {
			if gday <= day {
				waterFeed = gconv.String(float32(para.Fat_waterFeed[k]) / 10)
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_TIMER {
		waterFeed = gconv.String(float32(para.Tmr_waterFeed / 10))
	}

	return waterFeed
}

// 获取联动的控制配置(暂时全局)
func getLinkageConfig() (list gdb.List) {
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	sql := "SELECT * FROM b_dfxw_linkage_config ORDER BY `mode`,begin_day,begin_temp ASC;; "
	re, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
	} else {
		list = re.List()
	}
	return list
}

// 根据配置获取当前水料比偏移值
func getWaterFeedOffset(mode int, day int, temp float32) (offset float32) {
	db := g.DB(g.Cfg().GetString("custom.db_name"))
	sql := fmt.Sprintf("SELECT * FROM b_dfxw_linkage_config WHERE `mode`=%d AND begin_day<%d AND %d<=end_day AND IFNULL(begin_temp,-127)<%.2f AND %.2f<=IFNULL(end_temp,999);", mode, day, day, temp, temp)
	re, err := db.GetAll(sql)
	if err != nil {
		g.Log().Line().Error(err)
		offset = 0
	} else {
		list := re.List()
		if len(list) > 0 {
			item := list[0]
			offset = gconv.Float32(item["offset"])
		} else {
			offset = 0
		}
	}
	return offset
}

// 填写新的水料比到参数
func createParaByWaterFeed(para DMXFGW.NodePara, mode int, day int, waterFeed float32) (isok bool, new_para DMXFGW.NodePara) {
	isok = false
	new_para = para
	if mode == DMXFGW.FEED_MODE_PG {
		for k, gday := range para.Pg_gday {
			if int(gday) <= day {
				d := decimal.NewFromFloat32(waterFeed)
				cd := d.Mul(decimal.NewFromInt32(10)).String()
				new_para.Pg_waterFeed[k] = gconv.Uint8(cd)
				isok = true
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_LACTATION {
		for k, gday := range para.La_gday {
			if int(gday) <= day {
				d := decimal.NewFromFloat32(waterFeed)
				cd := d.Mul(decimal.NewFromInt32(10)).String()
				new_para.La_waterFeed[k] = gconv.Uint8(cd)
				isok = true
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_NURSE {
		for k, gday := range para.Nr_gday {
			if int(gday) <= day {
				d := decimal.NewFromFloat32(waterFeed)
				cd := d.Mul(decimal.NewFromInt32(10)).String()
				new_para.Nr_waterFeed[k] = gconv.Uint8(cd)
				isok = true
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_FAT {
		for k, gday := range para.Fat_gday {
			if int(gday) <= day {
				d := decimal.NewFromFloat32(waterFeed)
				cd := d.Mul(decimal.NewFromInt32(10)).String()
				new_para.Fat_waterFeed[k] = gconv.Uint8(cd)
				isok = true
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_TIMER {
		d := decimal.NewFromFloat32(waterFeed)
		cd := d.Mul(decimal.NewFromInt32(10)).String()
		new_para.Tmr_waterFeed = gconv.Uint8(cd)
		isok = true
	}
	return isok, new_para
}

// 获取需要修改的水料比参数数据结构位置
func getParaWaterFeedAddrOffset(para *DMXFGW.NodePara, mode int, day int) (water_feed_idx int) {

	if mode == DMXFGW.FEED_MODE_PG {
		for k, gday := range para.Pg_gday {
			if int(gday) <= day {
				water_feed_idx = k
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_LACTATION {
		for k, gday := range para.La_gday {
			if int(gday) <= day {
				water_feed_idx = k
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_NURSE {
		for k, gday := range para.Nr_gday {
			if int(gday) <= day {
				water_feed_idx = k
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_FAT {
		for k, gday := range para.Fat_gday {
			if int(gday) <= day {
				water_feed_idx = k
			} else {
				break
			}
		}
	} else if mode == DMXFGW.FEED_MODE_TIMER {
		water_feed_idx = 0
	}
	return water_feed_idx
}

// SyncLinkAgeWaterFeed 同步单个饲喂器环控联动
func SyncLinkAgeWaterFeed(addr string, tn int, mode int, water_feed_idx int, water_feed float32) (isok bool, res string) {
	enable := 1
	if 1 > 2 {
		isok = false
		res = fmt.Sprintf("终端:%s忙碌中,联动同步参数失败", addr)
	} else {
		//model.ProcessDeviceList.PushBack(addr)
		//defer RemoveProcessDeviceByAddr(addr)

		c := CreateMqttClient("cid_SyncLinkAgeFeed_"+grand.Digits(10), nil, nil)
		if token := c.Connect(); token.Wait() && token.Error() != nil {
			//panic(token.Error())
			g.Log().Line(true).Error(token.Error().Error())
		}
		topic_s := "S/DMXFGW/" + addr
		topic_p := "P/DMXFGW/" + addr

		var _begin int64
		if mode == DMXFGW.FEED_MODE_PG {
			_begin = common.GetStructFieldOffset(DMXFGW.NodePara{}, "Pg_waterFeed") + int64(water_feed_idx)
		} else if mode == DMXFGW.FEED_MODE_LACTATION {
			_begin = common.GetStructFieldOffset(DMXFGW.NodePara{}, "La_waterFeed") + int64(water_feed_idx)
		} else if mode == DMXFGW.FEED_MODE_NURSE {
			_begin = common.GetStructFieldOffset(DMXFGW.NodePara{}, "Nr_waterFeed") + int64(water_feed_idx)
		} else if mode == DMXFGW.FEED_MODE_FAT {
			_begin = common.GetStructFieldOffset(DMXFGW.NodePara{}, "Fat_waterFeed") + int64(water_feed_idx)
		} else if mode == DMXFGW.FEED_MODE_TIMER {
			_begin = common.GetStructFieldOffset(DMXFGW.NodePara{}, "Tmr_waterFeed")
		}

		d := decimal.NewFromFloat32(water_feed)
		cd := d.Mul(decimal.NewFromInt32(10)).String()
		var t_buff = gconv.Uint8(cd)

		var data []byte
		data = append(data, uint8(tn))
		data = append(data, uint8(enable))
		data = append(data, t_buff)
		//data = append(data, crc16_bytes...) //crc16
		size := uint32(len(data))
		msg := DMXFGW.CreateMsg(0x06, uint32(_begin), size, data)

		// 订阅主题
		var iscallback bool
		var pack *DMXFGW.Pack
		var err error
		//var mqtt_msg mqtt.Message
		if token := c.Subscribe(topic_p, 0, func(c mqtt.Client, msg mqtt.Message) {
			topic := msg.Topic()
			topic_arr := gstr.Split(topic, "/")
			addr := topic_arr[2]
			g.Log().Line(false).Println("DMXFGW recv from topic:" + msg.Topic())
			g.Log().Line(false).Println("DMXFGW recv from addr:" + addr)
			//g.Log().Line(false).Debugf("DMXFGW recv msg: % X", msg.Payload())
			//解析返回
			pack, err = DMXFGW.DecodePack(msg.Payload())
			if err == nil {
				Addr := gbinary.LeDecodeToUint16(pack.Addr[:])
				if pack.C == 0x86 && Addr == uint16(_begin) {
					iscallback = true
					//mqtt_msg = msg
				}
			} else {
				iscallback = true
			}

		}); token.Wait() && token.Error() != nil {
			fmt.Println(token.Error())
		}

		// 发布消息
		token := c.Publish(topic_s, 0, false, msg)
		token.Wait()

		for i := 0; i <= 200; i++ {
			if iscallback {
				//解析返回
				if err != nil {
					isok = false
					res = err.Error()
				} else {
					_tn := pack.Data[0]
					_res := pack.Data[1]
					if int(_tn) == tn {
						if _res == 0 {
							time.Sleep(1 * time.Second)
							//设置成功
							isok = true
							res = "ok"
						} else {
							isok = false
							res = "设置失败"
						}
					} else {
						isok = false
						res = "设置失败:tn错误"
					}

				}
				break
			} else {
				//继续等待
				time.Sleep(100 * time.Millisecond)
			}
		}

		c.Disconnect(250)
		if !iscallback {
			isok = false
			res = "请求超时,设备可能已经离线"
		}
	}
	return isok, res
}

// DFXW_getToken DFXW获取token
func DFXW_getToken() {
	//g.Log().Line(true).Info("DFXW_getToken")
	client := g.Client()
	//client.SetHeader("scope", "all")
	//client.SetHeader("grant_type", "client_credentials")
	//client.SetHeader("client_id", g.Cfg().GetString("ext_biz.dfxw_auth_client_id"))
	//client.SetHeader("client_secret", g.Cfg().GetString("ext_biz.dfxw_auth_client_secret"))
	timeout := g.Cfg().GetInt("ext_biz.dfxw_request_timeout")
	client.SetTimeout(time.Duration(timeout) * time.Second)
	params := g.Map{}
	params["scope"] = "all"
	params["grant_type"] = "client_credentials"
	params["client_id"] = g.Cfg().GetString("ext_biz.dfxw_auth_client_id")
	params["client_secret"] = g.Cfg().GetString("ext_biz.dfxw_auth_client_secret")
	url := g.Cfg().GetString("ext_biz.dfxw_auth_url")
	res, err := client.Post(url, params)
	if err != nil {
		g.Log().Line(true).Error(err)
	} else {
		res_str := res.ReadAllString()
		//g.Log().Line(true).Infof("res=%s", res_str)
		json, _ := gjson.DecodeToJson(res_str)
		res_map := json.GetMap(".")
		//g.Log().Line(true).Debug(res_map)
		model.DFXW_TOKEN = gconv.String(res_map["access_token"])
		model.DFXW_TOKEN_TYPE = gconv.String(res_map["token_type"])
		//g.Log().Line(true).Infof("DFXW_TOKEN=%s", model.DFXW_TOKEN)
		//g.Log().Line(true).Debugf("DFXW_TOKEN_TYPE=%s", model.DFXW_TOKEN_TYPE)

	}

}

func DFXW_getUnitBizData(id_arr []string) (list g.List, biz_time string, err error) {
	client := g.Client()
	timeout := g.Cfg().GetInt("ext_biz.dfxw_request_timeout")
	client.SetTimeout(time.Duration(timeout) * time.Second)
	client.SetHeader("Authorization", "Bearer "+model.DFXW_TOKEN)
	client.SetContentType("application/json")
	url := g.Cfg().GetString("ext_biz.dfxw_sync_unit_url")
	json_str, _ := gparser.VarToJsonString(id_arr)
	res, err := client.Post(url, json_str)
	if err != nil {
		g.Log().Line(true).Error(err)
	} else {
		res_str := res.ReadAllString()
		g.Log().Line(true).Debugf("res=%s", res_str)
		json, _ := gjson.DecodeToJson(res_str)
		res_map := json.GetMap(".")
		g.Log().Line(true).Debug(res_map)

		code := gconv.Int(res_map["code"])
		message := gconv.String(res_map["message"])
		result := gconv.String(res_map["result"])
		timestamp := gconv.String(res_map["timestamp"])
		if code == 200 {
			biz_time = gtime.NewFromTimeStamp(gconv.Int64(timestamp)).String()
			items_json, _ := gjson.DecodeToJson(result)
			items := items_json.GetMaps(".")
			list = items
			g.Log().Line(true).Debug(list)
		} else if code == 401 {
			if message == "Token失效" {
				model.DFXW_TOKEN = "" //Token失效
				//刷新token
				for i := 0; i < 5; i++ {
					if model.DFXW_TOKEN == "" {
						DFXW_getToken()
					} else {
						break
					}
				}
			}
		} else {
			err = errors.New(message)
		}

	}

	return list, biz_time, err
}

func DFXW_getBreedOrgToFarm() (str string, err error) {
	client := g.Client()
	timeout := g.Cfg().GetInt("ext_biz.dfxw_request_timeout")
	client.SetTimeout(time.Duration(timeout) * time.Second)
	client.SetHeader("Authorization", "Bearer "+model.DFXW_TOKEN)
	client.SetContentType("application/json")
	url := g.Cfg().GetString("ext_biz.dfxw_getBreedOrgToFarm_url")
	res, errs := client.Get(url)
	if errs != nil {
		g.Log().Line(true).Error(errs)
		err = errs
	} else {
		res_str := res.ReadAllString()
		g.Log().Line(true).Infof("res=%s", res_str)
		json, _ := gjson.DecodeToJson(res_str)
		res_map := json.GetMap(".")
		code := gconv.Int(res_map["code"])
		message := gconv.String(res_map["message"])
		result := gconv.String(res_map["result"])
		if code == 200 {
			str = result
		} else {
			err = errors.New(message)
		}
	}
	return str, err
}

// CheckAllDeliver 监测是否需要联动拉料
func CheckAllDeliver() {
	deliverList, err := dao.T_hzfgw_node_deliver.GetList()
	if err != nil {
		g.Log().Line(true).Error(err)
		return
	}

	if len(deliverList) == 0 {
		return
	}

	houseList, err := dao.T_house.GetList(fmt.Sprintf(" IFNULL(deliver_sn,'')!='' AND deliver_enable>0 "))
	if err != nil {
		g.Log().Line(true).Error(err)
		return
	}
	if len(houseList) == 0 {
		return
	}

	db := g.DB(g.Cfg().GetString("custom.db_name"))
	now := gtime.Now()
	for _, deliverItem := range deliverList {
		sync := gconv.Int(deliverItem["sync"])

		addr := gconv.String(deliverItem["addr"])
		sn := gconv.String(deliverItem["sn"])
		tn := gconv.Int(deliverItem["tn"])
		pre_act_time := gconv.String(deliverItem["pre_act_time"]) //上次启动时间
		_pre_act_time := gtime.NewFromStr(pre_act_time)           //上次启动时间

		if sync == 1 {
			g.Log().Line(true).Infof("拉料设备SN:%s,已经存在待拉料")
			//拉料
			DeliverFeed(addr, sn, tn, 180, 1, "拉料失败的重试")
			continue
		}

		//对应联动的单元(只要有一个单元满足条件，即可执行拉料)
		reason := ""
		needDeliver := false
		for _, house := range houseList {

			id := gconv.Int(house["id"])
			housename := gconv.String(house["housename"])
			deliver_sn := gconv.String(house["deliver_sn"])
			deliver_feed_node_max := gconv.String(house["deliver_feed_node_max"]) //单个上次启动后下料量阈值(kg)
			deliver_unit_max := gconv.String(house["deliver_unit_max"])           //单元上次启动后下料量阈值(kg)
			deliver_timeout := gconv.Int(house["deliver_timeout"])                //上次启动后超时阈值(分钟)

			if deliver_sn != sn {
				//匹配绑定了拉料设备sn的单元
				continue
			}

			//判断是否启动后超时未启动
			if _pre_act_time.Add(time.Duration(deliver_timeout) * time.Minute).Before(now) {
				//超时未拉
				reason = fmt.Sprintf("单元:%s,拉料设备SN:%s,超时未拉料，需启动拉料。上次拉料时间=%s,拉料超时时间=%d", housename, deliver_sn, pre_act_time, deliver_timeout)
				g.Log().Line(true).Info(reason)
				needDeliver = true
				break
			}

			//判断下料量是否超阈值
			deviceData, err := db.GetAll(fmt.Sprintf("SELECT * FROM v_device WHERE houseid=%d", id))
			if err != nil {
				g.Log().Line(true).Error(err)
				break
			}

			var device_addr string
			deviceList := deviceData.List()
			if len(deviceList) > 0 {
				deviceItem := deviceList[0]
				device_addr = gconv.String(deviceItem["device_addr"])
			} else {
				break
			}

			recordData, err := db.GetAll(fmt.Sprintf("SELECT sn,hnum,`mode`,amount FROM t_hzfgw_record WHERE addr='%s' AND rectime>='%s' AND `mode`<=10;", device_addr, pre_act_time))
			if err != nil {
				g.Log().Line(true).Error(err)
				break
			}
			recordList := recordData.List()

			var totalAmount float32
			for _, record := range recordList {
				amount := gconv.Float32(record["amount"])
				sn := gconv.String(record["sn"])
				hnum := gconv.String(record["hnum"])
				//mode := gconv.Int(record["mode"])
				totalAmount += amount
				if amount >= gconv.Float32(deliver_feed_node_max) {
					needDeliver = true
					reason = fmt.Sprintf("单元:%s,拉料设备SN:%s,单台下料量已经阈值,需启动拉料。超量饲喂器编码:%s,栏位号:%s,下料量:%.2f，阈值:%.2f", housename, deliver_sn, sn, hnum, amount, gconv.Float32(deliver_feed_node_max))
					g.Log().Line(true).Info(reason)
					break
				}
			}
			if needDeliver {
				break
			}

			if totalAmount >= gconv.Float32(deliver_unit_max) {
				needDeliver = true
				reason = fmt.Sprintf("单元:%s,拉料设备SN:%s,单元料量已经超限,需启动拉料,总下料量=%.2f,阈值=%.2f", housename, deliver_sn, totalAmount, gconv.Float32(deliver_unit_max))
				g.Log().Line(true).Info(reason)
				break
			}

		}

		if needDeliver {
			//拉料
			DeliverFeed(addr, sn, tn, 180, 1, reason)
		} else {
			//无需拉料
			g.Log().Line(true).Infof("终端:%s,拉料设备SN:%s,无需拉料", addr, sn)
		}
	}
}

// DeliverFeed 拉料动作
func DeliverFeed(addr string, sn string, tn int, timer int, _action int, reason string) (res_code int, res_content string) {

	//_action := 1
	//timer := 180 //3小时
	items, errs := dao.T_hzfgw_node_deliver.GetList(fmt.Sprintf("sn=%s", sn))
	if errs != nil {
		g.Log().Line(true).Error(errs)
		return
	}
	if len(items) == 0 {
		return
	}

	item := items[0]

	c := CreateMqttClient("cid_Hub_"+grand.Digits(10), nil, nil)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		//panic(token.Error())
		g.Log().Line(true).Error(token.Error().Error())
	}
	topic_s := "S/DMXFGW/" + addr
	topic_p := "P/DMXFGW/" + addr
	g.Log().Line(false).Debug(topic_s)

	var data []byte
	if _action > 0 {
		//拉料
		action_data := uint8(_action)
		timer_data := gbinary.LeEncodeUint16(uint16(timer))
		data = append(data, uint8(tn))
		data = append(data, action_data)
		data = append(data, 0x00) //保留字段
		data = append(data, timer_data...)
	} else {
		//停止拉料
		timer_data := gbinary.LeEncodeUint16(uint16(timer))
		data = append(data, uint8(tn))
		data = append(data, 0x00) //action
		data = append(data, 0x00) //保留字段
		data = append(data, timer_data...)
	}

	size := len(data)
	msg := DMXFGW.CreateMsg(03, 0x06, uint32(size), data)
	g.Log().Line(false).Debugf("DMXFGW send msg: % X", msg)

	// 订阅主题
	var iscallback bool
	var pack *DMXFGW.Pack
	var err error
	//var mqtt_msg mqtt.Message
	if token := c.Subscribe(topic_p, 0, func(c mqtt.Client, msg mqtt.Message) {
		topic := msg.Topic()
		topic_arr := gstr.Split(topic, "/")
		addr := topic_arr[2]
		g.Log().Line(false).Println("DMXFGW recv from topic:" + msg.Topic())
		g.Log().Line(false).Println("DMXFGW recv from addr:" + addr)
		g.Log().Line(false).Debugf("DMXFGW recv msg: % X", msg.Payload())
		//解析返回
		pack, err = DMXFGW.DecodePack(msg.Payload())
		if err == nil {
			Addr := gbinary.LeDecodeToUint16(pack.Addr[:])
			if pack.C == 0x83 && Addr == 0x06 {
				iscallback = true
				//mqtt_msg = msg
			}
		} else {
			iscallback = true
		}

	}); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
	}

	// 发布消息
	token := c.Publish(topic_s, 0, false, msg)
	token.Wait()

	for i := 0; i <= 200; i++ {
		if iscallback {
			//解析返回
			if err != nil {
				//失败
				item["sync"] = 1
				item["sync_res"] = 1
				item["sync_res_content"] = err.Error()
			} else {
				//成功
				item["sync"] = 0
				item["sync_res"] = 0
				item["pre_act_time"] = common.NowString()
				item["sync_res_content"] = "ok"
			}
			break
		} else {
			//继续等待
			time.Sleep(100 * time.Millisecond)
		}
	}
	c.Disconnect(250)
	if !iscallback {
		//失败
		item["sync"] = 1
		item["sync_res"] = 1
		item["sync_res_content"] = "终端超时,终端可能已离线"
	}

	item["sync_time"] = common.NowString()
	_, err = dao.T_hzfgw_node_deliver.Update(item)
	if err != nil {
		g.Log().Line(true).Error(errs)
	} else {
		g.Log().Line(true).Infof("终端:%s,拉料设备SN:%s 拉料成功", addr, sn)

		//写入报警通知
	}

	log := gdb.Map{
		"addr":     addr,
		"sn":       sn,
		"act_time": common.NowString(),
		"act_res":  item["sync_res_content"],
		"reason":   reason,
	}
	_, err = dao.B_dfxw_deliver_log.Save(log)
	if err != nil {
		g.Log().Line(true).Error(errs)
	}

	res_code = gconv.Int(item["sync"])
	res_content = gconv.String(item["sync_res_content"])
	return res_code, res_content
}

// CheckFeedNodeReplace 检测替换饲喂器

// 1 备份饲喂器参数, 清除旧饲喂器告警，清除旧饲喂器参数和状态 (备份&重置数据)
//
// 2 设置网关内TN位置的饲喂器SN为新的的饲喂器SN (注册新饲喂器)
//
// 3 网关降级06  (升级设备)
//
// 4 单独升级新饲喂器到最新版本  (升级设备)
//
// 5 网关升级到最新 (升级设备)
//
// 6 参数还原到新饲喂器 (还原参数)
//
// 7 完成
func CheckFeedNodeReplace() {
	g.Log().Line(true).Info("CheckFeedNodeReplace")
	//db := g.DB(g.Cfg().GetString("custom.db_name"))
	items, err := dao.T_hzfgw_node_replace.GetList("step>0 and step<7 order by id desc")
	if err != nil {
		g.Log().Line(true).Error(err)
		return
	}

	if len(items) > 0 {
		for _, item := range items {
			step := gconv.Int(item["step"])
			status := gconv.String(item["status"])

			old_sn := gconv.String(item["old_sn"])
			new_sn := gconv.String(item["new_sn"])
			addr := gconv.String(item["addr"])
			tn := gconv.Int(item["tn"])

			//新的饲喂器可能是别的地方拆了过来的，先找到旧饲喂器的之前的位置
			/*var addr string
			var tn int
			nodes, _ := dao.T_hzfgw_node_para.GetList(fmt.Sprintf(" sn='%s' ", old_sn))
			if len(nodes) > 0 {
				addr = gconv.String(nodes[0]["addr"])
				tn = gconv.Int(nodes[0]["tn"])
			} else {
				g.Log().Line(true).Errorf(" CheckFeedNodeReplace old sn=%s can not find addr & tn !", old_sn)
				continue
			}

			//查看新的饲喂器是否被使用过，被使用过，则清除他的告警，并清除他的参数和状态
			nodes, _ = dao.T_hzfgw_node_para.GetList(fmt.Sprintf(" sn='%s' ", new_sn))
			if len(nodes) > 0 {
				_, err = dao.T_hzfgw_node_para.DeleteNode(new_sn)
				if err != nil {
					g.Log().Line(true).Error(err)
				}
			}*/

			feed_plan_id := gconv.Int(item["feed_plan_id"])

			//已经完成备份&重置数据，开始设置新SN(包含重试)
			if (step == 1 && status == model.FeedNodeReplaceStatusDone) || (step == 2 && status == model.FeedNodeReplaceStatusErr) {
				isok, err := SetNodeSn(addr, tn, new_sn)
				if err != nil {
					g.Log().Line(true).Error(err)
					item["step"] = 2
					item["status"] = model.FeedNodeReplaceStatusErr
				} else {
					if isok {
						//设置成功
						item["step"] = 2
						item["status"] = model.FeedNodeReplaceStatusDone
					}
				}

				dao.T_hzfgw_node_replace.Save(item)
			}

			//已经完成设置新SN, 网关降级06(包含重试)
			if (step == 2 && status == model.FeedNodeReplaceStatusDone) || (step == 3 && status == model.FeedNodeReplaceStatusErr) {
				isok, err := UpdateNMSWQGW(addr, "upHZFGW_N0106")
				if err != nil {
					g.Log().Line(true).Error(err)
					item["step"] = 3
					item["status"] = model.FeedNodeReplaceStatusErr
				} else {
					if isok {
						//设置成功,升级处理中，后续持续根据网关版本判断是否完成升级
						item["step"] = 3
						item["status"] = model.FeedNodeReplaceStatusProcess
					}
				}

				dao.T_hzfgw_node_replace.Save(item)
			}

			//网关降级06处理中
			if step == 3 && status == model.FeedNodeReplaceStatusProcess {
				devices, err := dao.T_device.GetList(fmt.Sprintf(" device_addr='%s' ", addr))
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					if len(devices) > 0 {
						device := devices[0]
						alive := gconv.Int(device["alive"])
						version := gconv.String(device["version"])
						if alive > 0 && version == "HV:03.00 SV:01.00 RV:01.06" {
							//网关降级06 成功
							item["step"] = 3
							item["status"] = model.FeedNodeReplaceStatusDone
							dao.T_hzfgw_node_replace.Save(item)
						}
					}
				}
			}

			//网关降级06已经完成,升级新饲喂器(包含重试)
			if (step == 3 && status == model.FeedNodeReplaceStatusDone) || (step == 4 && status == model.FeedNodeReplaceStatusErr) {
				isok, err := UpdateNode(addr, new_sn, "FW_HZSWQ_NV2032")
				if err != nil {
					g.Log().Line(true).Error(err)
					item["step"] = 4
					item["status"] = model.FeedNodeReplaceStatusErr
				} else {
					if isok {
						//设置成功,升级处理中，后续持续饲喂器版本判断是否完成升级
						item["step"] = 4
						item["status"] = model.FeedNodeReplaceStatusProcess
					}
				}

				dao.T_hzfgw_node_replace.Save(item)
			}

			//升级新饲喂器处理中
			if step == 4 && status == model.FeedNodeReplaceStatusProcess {
				nodes, err := dao.T_hzfgw_node_status.GetList(fmt.Sprintf(" sn='%s' ", new_sn))
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					if len(nodes) > 0 {
						node := nodes[0]
						online := gconv.Int(node["online"])
						version := gconv.String(node["version"])
						if online > 0 {
							if version == "HV:02.00 SV:00.25" || version == "HV:02.00 SV:00.27" || version == "HV:02.00 SV:00.28" || version == "HV:02.00 SV:00.32" || version == "HV:02.00 SV:00.33" || version == "HV:04.00 SV:00.42" {
								//升级新饲喂器 成功
								item["step"] = 4
								item["status"] = model.FeedNodeReplaceStatusDone
								dao.T_hzfgw_node_replace.Save(item)
							}
						}
					}
				}
			}

			//饲喂器已经升级，升级网关到最新(包含重试)
			if (step == 4 && status == model.FeedNodeReplaceStatusDone) || (step == 5 && status == model.FeedNodeReplaceStatusErr) {
				isok, err := UpdateNMSWQGW(addr, "upHZFGW_N0200_0001")
				if err != nil {
					g.Log().Line(true).Error(err)
					item["step"] = 5
					item["status"] = model.FeedNodeReplaceStatusErr
				} else {
					if isok {
						//设置成功,升级处理中，后续持续根据网关版本判断是否完成升级
						item["step"] = 5
						item["status"] = model.FeedNodeReplaceStatusProcess
					}
				}

				dao.T_hzfgw_node_replace.Save(item)
			}

			//升级网关到最新处理中
			if step == 5 && status == model.FeedNodeReplaceStatusProcess {
				devices, err := dao.T_device.GetList(fmt.Sprintf(" device_addr='%s' ", addr))
				if err != nil {
					g.Log().Line(true).Error(err)
				} else {
					if len(devices) > 0 {
						device := devices[0]
						alive := gconv.Int(device["alive"])
						version := gconv.String(device["version"])
						if alive > 0 {
							if version == "HV:03.00 SV:01.00 RV:01.10" || version == "HV:03.00 SV:02.00 RV:00.01" || version == "HV:03.00 SV:01.00 RV:01.06" {
								//升级网关 成功
								item["step"] = 5
								item["status"] = model.FeedNodeReplaceStatusDone
								dao.T_hzfgw_node_replace.Save(item)
							}
						}
					}
				}
			}

			//网关已经升级，还原参数(包含重试)
			if (step == 5 && status == model.FeedNodeReplaceStatusDone) || (step == 6 && status == model.FeedNodeReplaceStatusErr) {
				params := redis.GetGwNodeBakupPara(addr, old_sn)
				if params == "" {
					//无备份参数，不还原,正常结束
					item["step"] = 7
					item["status"] = model.FeedNodeReplaceStatusDone
				} else {
					fp := DMXFGW.FNodePara{}
					err = gjson.DecodeTo(params, &fp)
					if err != nil {
						g.Log().Line(true).Error(err)
						continue
					}
					fp.Sn = new_sn

					str, _ := gparser.VarToJsonString(fp)

					isok, err := SetNodeParam(addr, tn, str, feed_plan_id)
					if err != nil {
						g.Log().Line(true).Error(err)
						item["step"] = 6
						item["status"] = model.FeedNodeReplaceStatusErr
					} else {
						if isok {
							//所有完成
							item["step"] = 7
							item["status"] = model.FeedNodeReplaceStatusDone
						}
					}
				}
				dao.T_hzfgw_node_replace.Save(item)
			}
		}
	}
}

func ClearRedisLastRid() {

}
